package com.crazymakercircle.dhptester.distributed;

import com.crazymakercircle.constants.ServerConstants;
import com.crazymakercircle.dhptester.benchmark.HttpBenchmark;
import com.crazymakercircle.entity.TaskData;
import com.crazymakercircle.entity.TestResultData;
import com.crazymakercircle.util.JsonUtil;
import com.crazymakercircle.util.ObjectUtil;
import com.crazymakercircle.util.ThreadUtil;
import com.crazymakercircle.zk.CuratorZKclient;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;

import java.util.concurrent.TimeUnit;

/**
 * create by 尼恩 @ 疯狂创客圈
 **/
@Data
@Slf4j
public class TaskExecutor
{
    //Zk客户端
    private CuratorFramework client = null;

    private String pathRegistered = null;
    private TaskData taskData = null;


    private static TaskExecutor singleton = null;
    private static final String path = ServerConstants.MANAGE_PATH;


    public synchronized static TaskExecutor getSingleton()
    {
        if (null == singleton)
        {
            singleton = new TaskExecutor();
            singleton.init();
        }
        return singleton;
    }

    private TaskExecutor()
    {

    }

    private boolean inited = false;
    private boolean justStart = true;

    /**
     * 初始化节点管理
     */
    public synchronized void init()
    {

        if (inited)
        {
            return;
        }
        inited = true;

        if (null == client)
        {
            this.client = CuratorZKclient.getSingleton().getClient();

        }

    }

    public void startMoniter()
    {

        ThreadUtil.delayRun(() -> justStart = false, 5, TimeUnit.SECONDS);

        try
        {
            //订阅节点的增加和删除事件

            String monitor_path = ServerConstants.TASK_PATH;
            PathChildrenCache childrenCache = new PathChildrenCache(client, monitor_path, true);
            PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener()
            {

                @Override
                public void childEvent(CuratorFramework client,
                                       PathChildrenCacheEvent event) throws Exception
                {
                    log.info("开始监听其他的ImWorker子节点:-----");
                    ChildData data = event.getData();
                    switch (event.getType())
                    {
                        case CHILD_ADDED:
                            log.info("CHILD_ADDED : " + data.getPath() + "  数据:" + data.getData());
                            processNodeAdded(data);
                            break;
                        case CHILD_REMOVED:
                            log.info("CHILD_REMOVED : " + data.getPath() + "  数据:" + data.getData());
                            processNodeRemoved(data);
                            break;
                        case CHILD_UPDATED:
                            log.info("CHILD_UPDATED : " + data.getPath() + "  数据:" + new String(data.getData()));
                            break;
                        default:
                            log.debug("[PathChildrenCache]节点数据为空, path={}", data == null ? "null" : data.getPath());
                            break;
                    }

                }

            };

            childrenCache.getListenable().addListener(
                    childrenCacheListener, ThreadUtil.getIoIntenseTargetThreadPool());
            System.out.println("Register zk watcher successfully!");
            childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

        } catch (Exception e)
        {
            e.printStackTrace();
        }
    }

    private void processNodeRemoved(ChildData data)
    {

        byte[] payload = data.getData();
        TaskData node = ObjectUtil.JsonBytes2Object(payload, TaskData.class);

        String id = TaskData.getRunIdByPath(data.getPath());
        log.info("[TreeCache]节点删除, path={}, data={}",
                data.getPath(), JsonUtil.pojoToJson(node));


    }

    /**
     * 节点增加的处理
     *
     * @param data 新节点
     */
    private void processNodeAdded(ChildData data)
    {
        byte[] payload = data.getData();

        TaskData taskData = TaskData.fromJson(payload);

        if(taskData==null)
        {
            return;
        }
        String id = TaskData.getRunIdByPath(data.getPath());
        taskData.setRunId(id);

        log.info("[TreeCache]节点更新端口, path={}, data={}",
                data.getPath(), JsonUtil.pojoToJson(taskData));

        if (justStart) return;

        if (WorkerNode.isBusy)
        {
            System.out.println("isbusy = " + WorkerNode.isBusy);
            return;
        }
        WorkerNode.isBusy = true;

        HttpBenchmark httpBenchmark = new HttpBenchmark();

        TestResultData resultData = httpBenchmark.runJsonTestCase(taskData.getJson());

        if (null == resultData)
        {
            System.out.println(" error occur !  ");
            return;
        }

        String resultPath = data.getPath() + "/node-" + WorkerNode.getSingleton().getId();
        System.out.println("resultPath = " + resultPath);

        if (!CuratorZKclient.getSingleton().isNodeExist(resultPath))
        {
            CuratorZKclient.getSingleton().createNode(resultPath, resultData.toJson());
        } else
        {
            CuratorZKclient.getSingleton().updateData(resultPath, resultData.toJson());

        }
        WorkerNode.isBusy = false;

    }


}
